package com.ficus.road.maintain.grpc;

import com.ficus.road.maintain.core.model.Car;
import com.ficus.road.maintain.core.model.GpsHistory;
import com.ficus.road.maintain.core.model.GpsLocation;
import com.ficus.road.maintain.service.CarService;
import com.ficus.road.maintain.service.DataProcessorManager;
import com.ficus.road.maintain.service.GpsHistoryService;
import com.ficus.road.maintain.util.AliyunOSSUtil;
import com.ficus.road.maintain.util.BatchUtil;
import com.ficus.road.maintain.util.queue.BlockingQueueMonitorDecorator;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import ficus.car_demo.FrameMeta;
import ficus.car_demo.SenderGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.RandomUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.BiFunction;


@GrpcService
@Slf4j
public class FrameMetaGrpcServer extends SenderGrpc.SenderImplBase {
    private static final String SUCCESS_MESSAGE = "success";
    public static BlockingQueue<FrameMeta.FrameResult> frameResultQueue = new BlockingQueueMonitorDecorator<>(
            "send_frame_result_queue",
            new ArrayBlockingQueue<>(1500));
    ThreadPoolExecutor saveGpsExecutor = new ThreadPoolExecutor(1, 1,
            10L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(200), new CustomizableThreadFactory("saveGpsExecutor-pool-"));
    ThreadPoolExecutor saveGpsMinuteExecutor = new ThreadPoolExecutor(1, 1,
            10L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(200), new CustomizableThreadFactory("saveGpsMinuteExecutor-pool-"));
    volatile Map<String, Long> carGpsMap = Maps.newConcurrentMap();
    volatile Map<String, Long> carGpsRetryMap = Maps.newConcurrentMap();
    volatile Map<String, Long> carGpsMinuteMap = Maps.newConcurrentMap();
    @Autowired
    private DataProcessorManager dataProcessorManager;
    @Autowired
    private CarService carService;
    @Autowired
    private GpsHistoryService gpsHistoryService;
    private Map<String, FrameMetaProcessor> frameMetaProcessorMap = new ConcurrentHashMap<>();
    @Value("${history-video.prefix}")
    private String prefix;
    @Value("${hit-config.gps-save-duration}")
    private int gpsSaveDuration;
    @Value("${hit-config.gps-minute-save-duration}")
    private int gpsMinuteSaveDuration;
    @Autowired
    private AliyunOSSUtil aliyunOSSUtil;
    private BiFunction<String, FrameMeta.FrameResult, CompletableFuture<Void>> asyncFileLengthFunc;
//    @Value("${myGrpc.test}")
//    private Boolean flag;

    @PostConstruct
    public void init() throws IOException {
        BiFunction<String, List<FrameMeta.FrameResult>, List<Void>> batchFunc =
                (key, inputs) -> {
                    sendFrameResult(inputs);
                    List<Void> results = new ArrayList<>(inputs.size());
                    for (int i = 0; i < results.size(); i++) {
                        results.add(null);
                    }
                    return results;
                };
        BiFunction<String, List<FrameMeta.FrameResult>, Boolean> batchChecker =
                (key, inputs) -> {
                    if (inputs.size() > 250) {
                        return true;
                    }
                    return false;
                };
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1000);
        asyncFileLengthFunc = BatchUtil.createBatchFunc(
                batchFunc, batchChecker, 10000, new ThreadPoolExecutor(
                        2, 2,
                        1, TimeUnit.SECONDS, queue, new CustomizableThreadFactory("internalBatchExecutor-pool-")));

        new Thread(this::consumeQueue).start();
    }

    public void consumeQueue() {
        while (true) {
            try {
                FrameMeta.FrameResult frameResult = frameResultQueue.take();
                asyncFileLengthFunc.apply(frameResult.getCarId(), frameResult);
            } catch (Throwable e) {
                log.error("consume frame_result_queue met exception:", e);
            }
        }
    }

    /**
     * 每天凌晨清空map
     */
    @Scheduled(cron = "0 0 0 * * ?")
    void clearTimestampMap() {
        carGpsMap.clear();
        carGpsMinuteMap.clear();
        carGpsRetryMap.clear();
    }

    public void sendFrameResult(List<FrameMeta.FrameResult> list) {
        if (CollectionUtils.isNotEmpty(list)) {
            String carId = list.get(0).getCarId();
            FrameMetaProcessor frameMetaProcessor = frameMetaProcessorMap.get(carId);
            if (frameMetaProcessor == null) {
                frameMetaProcessor = new FrameMetaProcessor(prefix, aliyunOSSUtil, dataProcessorManager);
                frameMetaProcessorMap.put(carId, frameMetaProcessor);
                //实时数据处理
                new Thread(frameMetaProcessor).start();
            }
            try {
                frameMetaProcessor.put(list);
            } catch (InterruptedException e) {
                log.error("InterruptedException", e);
            }
        }
    }

    // 旧版
    @Override
    public void sendMeta(FrameMeta.SendRequest request, StreamObserver<FrameMeta.SendReply> responseObserver) {
//        dataProcessorManager.put(request);
// 往云端发送数据，发送失败抛异常
        List<FrameMeta.FrameResult> framesList = request.getFramesList();
        for (FrameMeta.FrameResult frameResult : framesList) {
            boolean offer = frameResultQueue.offer(frameResult);
            if (!offer) {
                log.warn("add frame into queue error: {}", frameResult.getSequenceId());
                frameResultQueue.poll();
                frameResultQueue.offer(frameResult);
            }
        }
        FrameMeta.SendReply reply = FrameMeta.SendReply.newBuilder().setMessage(SUCCESS_MESSAGE).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void registerCar(FrameMeta.CarRegisterRequest request, StreamObserver<FrameMeta.CarRegisterReply> responseObserver) {
        Car car = Car.builder()
                .id(request.getCarId())
                .plateNo(request.getPlateNo())
                .build();
        carService.registerCar(car);
        FrameMeta.CarRegisterReply reply = FrameMeta.CarRegisterReply.newBuilder().setCarId(request.getCarId()).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    //新版 sapphire调用该方法
    @Override
    public void sendMetaInfo(FrameMeta.SendMetaRequest request, StreamObserver<FrameMeta.SendReply> responseObserver) {
        dataProcessorManager.put(request);
        FrameMeta.SendReply reply = FrameMeta.SendReply.newBuilder().setMessage(SUCCESS_MESSAGE).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }
    public void sendMetaInfo1(FrameMeta.SendMetaRequest request) {
        dataProcessorManager.put(request);
    }

    // 存gps轨迹
    @Override
    public void sendGpsRequest(FrameMeta.GpsListRequest request, StreamObserver<FrameMeta.SendReply> responseObserver) {
//        log.info("receive gps information");
//        if (flag) {
//            try {
//                Thread.sleep(8000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//        }
        //记录gps信息
        saveGpsExecutor.submit(() -> saveGps(request, Boolean.FALSE));
        //记录分钟级别gps信息
        saveGpsMinuteExecutor.submit(() -> saveMinuteGps(request));
        FrameMeta.SendReply reply = FrameMeta.SendReply.newBuilder().setMessage(SUCCESS_MESSAGE).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    public void sendGpsRequest1(FrameMeta.GpsListRequest request) {
        //记录gps信息
        saveGpsExecutor.submit(() -> saveGps(request, Boolean.FALSE));
        //记录分钟级别gps信息
        saveGpsMinuteExecutor.submit(() -> saveMinuteGps(request));
    }

    // 接收重传的gps轨迹
    @Override
    public void sendGpsRequestRetry(FrameMeta.GpsListRequest request, StreamObserver<FrameMeta.SendReply> responseObserver) {
        //记录gps信息
        saveGpsExecutor.submit(() -> saveGps(request, Boolean.TRUE));
        //记录分钟级别gps信息
        saveGpsMinuteExecutor.submit(() -> saveMinuteGps(request));
        if (CollectionUtils.isNotEmpty(request.getGpsRequestList())) {
            log.info("接收重传的gps轨迹, carId: {}", request.getGpsRequestList().get(0).getCarId());
        }
        FrameMeta.SendReply reply = FrameMeta.SendReply.newBuilder().setMessage(SUCCESS_MESSAGE).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    private void saveGps(FrameMeta.GpsListRequest request, Boolean isRetry) {
        request.getGpsRequestList().forEach(gpsRequest -> {
            String carId = gpsRequest.getCarId();
            FrameMeta.GPSResult gpsResult = gpsRequest.getGpsResult();
            long timestamp = gpsRequest.getTimestamp() / 1000;
            if (isRetry) {
                if (carGpsRetryMap.containsKey(carId)) {
                    Long lastGpsTime = carGpsRetryMap.get(carId);
                    //这个方法的GPS记录保存时间间隔是2秒
                    if (timestamp - lastGpsTime < gpsSaveDuration) {
                        log.debug("time is not enough, time={}, lastTime={}, duration={}", timestamp, lastGpsTime, gpsSaveDuration);
                        return;
                    }
                }
            } else {
                if (carGpsMap.containsKey(carId)) {
                    Long lastGpsTime = carGpsMap.get(carId);
                    //这个方法的GPS记录保存时间间隔是2秒
                    if (timestamp - lastGpsTime < gpsSaveDuration) {
                        log.debug("time is not enough, time={}, lastTime={}, duration={}", timestamp, lastGpsTime, gpsSaveDuration);
                        return;
                    }
                }
            }
            // 过滤(0, 0)数据
            if (gpsRequest.getGpsResult().getLatitude() > 1) {
                // 苟一下
                int speed = (int) (gpsResult.getSpeed() * 3.6);
                if ("YTKJ20210076".equals(carId)) {
                    speed = RandomUtils.nextInt(10, 70);
                }
                gpsHistoryService.create(GpsHistory.builder()
                        .id(carId + timestamp)
                        .longitude(gpsResult.getLongitude())
                        .latitude(gpsResult.getLatitude())
                        .altitude(gpsResult.getHeightMsl())
                        .location(Lists.newArrayList(gpsResult.getLongitude(), gpsResult.getLatitude()))
                        .timestamp(timestamp)
                        .speed(speed)
                        .direction((int) gpsResult.getHeading())
                        .creationTime(new Date())
                        .carId(carId)
                        .isFixed(Boolean.FALSE)
                        .isManualFixed(Boolean.FALSE)
                        .isRetry(isRetry)
                        .build());
                if (isRetry) {
                    carGpsRetryMap.put(carId, timestamp);
                } else {
                    carGpsMap.put(carId, timestamp);
                }
                GpsLocation gpsLocation = GpsLocation.builder().longitude(gpsResult.getLongitude())
                        .latitude(gpsResult.getLatitude()).build();
                carService.updateCarLocation(carId, gpsLocation, (int) (gpsResult.getSpeed() * 3.6), (int) gpsResult.getHeading());
            }else {
                log.error("gps error,getLatitude =0, car={}", carId);
            }
        });
    }

    private void saveMinuteGps(FrameMeta.GpsListRequest request) {
        request.getGpsRequestList().forEach(gpsRequest -> {
            String carId = gpsRequest.getCarId();
            FrameMeta.GPSResult gpsResult = gpsRequest.getGpsResult();
            long timestamp = gpsRequest.getTimestamp() / 1000;
            if (carGpsMinuteMap.containsKey(carId)) {
                Long lastGpsTime = carGpsMinuteMap.get(carId);
                // TIME_MINUTE_DURATION_DEFAULT在BaseProcessor中通过配置文件设置，时间间隔小于一分钟，则直接返回
                if (timestamp - lastGpsTime < gpsMinuteSaveDuration) {
                    log.debug("time is not enough, time={}, lastTime={}, duration={}", timestamp, lastGpsTime, gpsMinuteSaveDuration);
                    return;
                }
            }
            if (gpsResult.getLatitude() > 1) {
                //时间间隔超过一分钟，则创建新的分钟级别的GPS记录
                int speed = (int) (gpsResult.getSpeed() * 3.6);
//                if ("YTKJ20210076".equals(carId)) {
//                    speed = RandomUtils.nextInt(10, 70);
//                }
                gpsHistoryService.createMinute(GpsHistory.builder()
                        .id(carId + timestamp)
                        .longitude(gpsResult.getLongitude())
                        .latitude(gpsResult.getLatitude())
                        .altitude(gpsResult.getHeightMsl())
                        .location(Lists.newArrayList(gpsResult.getLongitude(), gpsResult.getLatitude()))
                        .timestamp(timestamp)
                        .speed(speed)
                        .direction((int) gpsResult.getHeading())
                        .creationTime(new Date())
                        .carId(carId)
                        .build());
                //更新map中的时间戳
                carGpsMinuteMap.put(carId, timestamp);
            }
        });
    }
}